package com.xbai.spark.recall.engine

import com.xbai.spark.recall.data.{DataLoader, Perprocessing}
import com.xbai.spark.recall.engine.training.UserCFTraining
import org.apache.spark.sql.functions.desc
import org.apache.spark.sql.{DataFrame, SparkSession, functions}

/**
  * @author xbai
  * @Date 2021/1/14
  */
class UserCFEngine(spark: SparkSession) {

  var userSimilarCache: DataFrame = _
  var userItemRatingCache: DataFrame = _

  def init(): Unit = {
    val userItemRating: DataFrame = createUserItemRating()
    new UserCFTraining().trainingEngine(userItemRating, spark)
    userSimilarCache = spark.sql("select * from userSimilar")
    userItemRatingCache = spark.sql("select * from userItemRating")
  }

  /**
    * 获取用户最相似的用户
    * @param findUserId 用户id
    * @param num 最相似的数量
    * @return 用户列表
    */
  def getSimilarUsers(findUserId: String, num: Int): DataFrame = {
    val users: DataFrame = userSimilarCache.filter(functions.col("userId").equalTo(findUserId))
      .sort(desc("usersim"))
      .limit(num)
    println("========== users ==========")
    users.show(3)
    users
  }

  /**
    * 获取相似用户的物品
    * @param users 相似用户集合
    * @return
    */
  def getItemsBySimilarUser(users: DataFrame): DataFrame = {
    getItemsBySimilarUser(users, userItemRatingCache)
  }

  /**
    * 获取相似用户的物品
    * @param users 相似用户集合
    * @param userItemRating user-item-rating
    * @return
    */
  def getItemsBySimilarUser(users: DataFrame, userItemRating: DataFrame): DataFrame = {
    val userIds: Array[String] = users.selectExpr("userId1").collect().map(_.mkString)
    userIds.foreach(println)
    var items: DataFrame = null
    userIds.foreach(userId => {
      val temp: DataFrame = userItemRating.filter(functions.col("userId").equalTo(userId))
        .sort(desc("rating"))
        .limit(40)
        .select("itemId")
      if (items == null) {
        items = temp
      } else {
        items = items.union(temp)
      }
    })
    println("========== itemsSimilar ==========")
    items.show(3)
    items.distinct()
  }

  /**
    * 创建用户物品评分
    * @return user-item-rating
    */
  def createUserItemRating(): DataFrame = {
    val dataLoader = new DataLoader
    val baseData: DataFrame = dataLoader.getUserRemainFromDB(spark)
    val perProcessing = new Perprocessing
    val userItemRating: DataFrame = perProcessing.createItemScoreByUser(baseData)
    println("========== userItemRating ==========")
    userItemRating.show(3)
    userItemRating
  }

}
